/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.listener;

import java.io.*;
import java.sql.*;
import java.util.concurrent.Semaphore;
import java.net.*;
import java.security.*;

import nfse.DiskManager;
import nfse.MetaDatabase;
import nfse.NetFSE;
import nfse.NetFSEUtil;
import nfse.SensorInfo;

public class NetListenerSocket extends Thread {

    private final Semaphore available = new Semaphore(1, true);

    private Socket sock;

    private NetListenerServer parent;

    private String dataDir = "";

    private int sensorID = -1;

    private int typeID = -1;

    private Statement s = null;

    DataOutputStream dos2 = null;

    private long lastCalled = -1;

    MetaDatabase mdb = null;

    String path = "";

    MessageDigest md = null;

    boolean computeChecksum = false;

    long currentCount = 0;

    public NetListenerSocket(Socket s, NetListenerServer parentServer, boolean computeChecksum) {
        sock = s;
        this.parent = parentServer;
        this.computeChecksum = computeChecksum;
    }

    public NetListenerSocket(Socket s, NetListenerServer parentServer) {
        sock = s;
        this.parent = parentServer;
        this.computeChecksum = true;
    }

    /***************************************************************************
     * This method reads in the configuration for the data server.
     * 
     */
    private void processConfig() {
        LineNumberReader in = null;
        try {
            File conf = new File(NetFSE.CONFIG_FILE);
            if (!conf.exists()) {
                return;
            }

            in = new LineNumberReader(new FileReader(conf));
            String line = in.readLine();
            while (line != null) {
                if (line.startsWith("NetListener::datadir ")) {
                    dataDir = line.substring(line.indexOf(' ')).trim();
                    if (!dataDir.endsWith("/"))
                        dataDir += "/";
                    File temp = new File(dataDir);
                    if (!temp.exists())
                        temp.mkdirs();
                }
                line = in.readLine();
            }
            in.close();

        } catch (Exception E) {
            E.printStackTrace();
            try {
                in.close();
            } catch (IOException ioe) {
                ioe.printStackTrace();
            }
        }
    }

    public long getLastCalled() {
        return lastCalled;
    }

    public long getCurrentCount() {
        return currentCount;
    }

    public void run() {
        try {

            processConfig();

            long startMS = System.currentTimeMillis();
            DataInputStream dis = new DataInputStream(sock.getInputStream());
            DataOutputStream dos = new DataOutputStream(sock.getOutputStream());
            this.sensorID = dis.readInt();
            
            SensorInfo sensorInfo = NetFSE.getSensorInfo(sensorID);
            if (sensorInfo == null) {
                System.err.println("Unknown sensor ID specified: " + sensorID + ".");
                dos.writeInt(-1);
                dos.flush();
                dos.close();
                dis.close();
                sock.close();
                return;
            } else {
                dos.writeInt(0);
            }

            this.typeID = sensorInfo.getToolID();

            mdb = MetaDatabase.create();
            s = mdb.createStatement();

            nextFile();

            int len = dis.readInt();

            long ts = System.currentTimeMillis();

            // 
            long count = 0;
            // long currentCount = 0;
            while (len != -1) {
                try {
                    // if (len > 165)
                    // System.out.println("Len: " + len);

                    int bytesRead = 0;
                    byte[] bytes = new byte[len];

                    /***********************************************************
                     * Read bytes from the socket connection into an array
                     * (bytes) of length len until the specified length is met.
                     */
                    while (bytesRead < len)
                        bytesRead += dis.read(bytes, bytesRead, len - bytesRead);

                    if (bytesRead != len) {
                        throw (new Exception("Net/Listener: Bad data, not enough bytes."));
                    }

                    try {
                        // System.out.println(new String(bytes));
                        writeRecord(bytes);
                    } catch (Exception e) {
                        e.printStackTrace();
                        len = -1;
                    }
                    if (len != -1) {
                        len = dis.readInt();
                    }
                    count++;
                    currentCount++;
                    if (count % 10000 == 0) {
                        // Output statistics periodically
                        long ts2 = System.currentTimeMillis();
                        long ms = ts2 - ts;
                        long elapsedMS = ts2 - startMS;
                        float elapsedSec = (float) elapsedMS / (float) 1000;
                        float sec = (float) ms / (float) 1000;
                        float sps1 = (float) count / elapsedSec;
                        float sps2 = (float) 10000 / sec;
                        System.out.println("Net/Listener: Total Records=" + count + "; Current RPS= " + sps2
                                + "; Average RPS= " + sps1);
                        ts = System.currentTimeMillis();

                        double dataDirFullPercent = NetFSE.getDiskFullPercent(dataDir);

                        while (dataDirFullPercent > 90.0) {
                            System.out.println("Net/Listener: Almost out of disk space on " + dataDir
                                    + ". Waiting for DiskManager to free up space.");
                            dataDirFullPercent = DiskManager.reclaimSpaceFromDataFiles(dataDir);
                            Thread.sleep(1000);
                        }
                    }

                    if (count % 100000 == 0) {
                        closeFile(currentCount);
                        currentCount = 0;
                        nextFile();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    len = -1;
                }
            }

            closeFile(currentCount);

            System.out.println("Net/Listener: Processed " + count + " records.");

            try {
                dos.writeInt(-1);
                dos.flush();
            } catch (Exception e) {

            }
            try {
                dis.close();
            } catch (Exception e) {

            }
            try {
                dos.close();
            } catch (Exception e) {

            }
            try {
                dos2.flush();
                dos2.close();
            } catch (Exception e) {

            }
            try {
                sock.close();
            } catch (Exception e) {

            }

            s.close();
            mdb.closeConnection();

            System.out.println("Net/Listener socket exited.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void closeFile(long count) throws Exception {
        try {
            available.acquire();
            String sql = "";
            if (path.length() > 0) {
                try {
                    dos2.flush();
                    dos2.close();
                } catch (IOException ioe) {
                    ioe.printStackTrace();
                }
                String digest = "";
                if (computeChecksum)
                    digest = NetFSEUtil.convertToHex(new String(md.digest()));
                // System.out.println(digest);
                sql = "update data_files set Records=" + count + ", Status=1, Digest='" + digest + "' where Path='"
                        + path + "'";
                s.execute(sql);
            }
        } finally {
            available.release();
        }
    }

    public String nextFile() throws Exception {
        try {
            available.acquire();

            lastCalled = System.currentTimeMillis();
            long maxID = -1;
            String sql = "select max(file_id) from data_files";
            ResultSet rs = s.executeQuery(sql);
            if (rs.next()) {
                maxID = rs.getLong(1);
            }
            rs.close();

            long fileID = -1;
            do {
                sql = "select nextval('file_id_seq')";
                rs = s.executeQuery(sql);
                rs.next();
                fileID = rs.getLong(1);
                rs.close();
            } while (fileID <= maxID);

            path = dataDir;
            File f = new File(path);
            if (!f.exists())
                f.mkdir();
            path += NetFSE.NODE_ID + "/";
            f = new File(path);
            if (!f.exists())
                f.mkdir();
            path += sensorID + "/";
            f = new File(path);
            if (!f.exists())
                f.mkdir();

            long partition = -1;
            if (parent.getPartitionSize() != -1) {
                partition = fileID / parent.getPartitionSize();
                partition++;
                path += partition + "/";
                f = new File(path);
                if (!f.exists())
                    f.mkdirs();
            }

            path += fileID + ".data";

            // System.out.println("New file: '" + path + "'");

            mdb.addDataFile(fileID, typeID, sensorID, NetFSE.NODE_ID, path);
            
            FileOutputStream fOut = new FileOutputStream(path);
            dos2 = new DataOutputStream(fOut);

            if (computeChecksum) {
                try {
                    md = MessageDigest.getInstance("SHA");
                } catch (Exception e) {
                    e.printStackTrace();
                    return "";
                }
            }
            return path;
        } finally {
            available.release();
        }
    }

    private void writeRecord(byte[] bytes) throws Exception {
        try {
            available.acquire();
            dos2.writeInt(bytes.length);
            dos2.write(bytes);
            if (computeChecksum) {
                try {
                    md.update(new Integer(bytes.length).byteValue());
                    md.update(bytes);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // System.out.println(new String(bytes));
        } finally {
            available.release();
        }
    }

}